package com.dai.rpc.netty.handler.idle;

import com.dai.rpc.exceptions.MyRpcException;
import com.dai.rpc.netty.cache.ChannelCache;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.util.Timeout;
import io.netty.util.Timer;
import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

@ChannelHandler.Sharable//用来说明ChannelHandler是否可以在多个channel直接共享使用
@Slf4j
public abstract class ConnectionWatchdog extends ChannelInboundHandlerAdapter implements TimerTask,ChannelHandlerHolder,CacheClearHandler{

    private Bootstrap bootstrap;
    private CompletableFuture<Channel> channelCompletableFuture;
    private Timer timer;
    private InetSocketAddress netAddress;
    private boolean reconnection;
    private int reConnecteNum;
    private  ChannelCache channelCache;

    public ConnectionWatchdog(Bootstrap bootstrap, CompletableFuture<Channel> channelCompletableFuture, Timer timer, InetSocketAddress netAddress, boolean reconnection, ChannelCache channelCache) {
        this.bootstrap = bootstrap;
        this.channelCompletableFuture = channelCompletableFuture;
        this.timer = timer;
        this.netAddress = netAddress;
        this.reconnection = reconnection;
        this.channelCache = channelCache;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 成功连接
        log.info("链路已经激活，重试次数清0");
        reConnecteNum = 0;
        // 标识通道状态
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("链路关闭");
        // 重试连接
        if(reconnection){
            log.info("链路关闭，进行重连，当前重连次数为：{}",reConnecteNum);
            if(reConnecteNum < 12){
                reConnecteNum++;
            }else{
                reconnection = false;
                // 删除服务提供端缓存
                clear(netAddress);
            }

            int timeout = 2 << reConnecteNum;
            timer.newTimeout(this, timeout, TimeUnit.MILLISECONDS);
        }
        // 标识状态
        ctx.fireChannelInactive();

    }

    @Override
    public void run(Timeout timeout) throws Exception {
        // 执行定时任务,重新连接
        ChannelFuture connect = null;
        synchronized (bootstrap){
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    socketChannel.pipeline().addLast(channelHandlers());
                }
            });
            connect = bootstrap.connect(netAddress);
        }

        connect.addListener(new ChannelFutureListener() {

            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                if(channelFuture.isSuccess()){
                    // 代表连接成功，将channel放入任务中
                    channelCompletableFuture.complete(channelFuture.channel());
                    // 放入channel缓存
                    channelCache.set(netAddress,channelFuture.channel());
                }else{
                    // 连接失败，再次尝试重连
                    channelCompletableFuture.completeExceptionally(channelFuture.cause());
                    channelFuture.channel().pipeline().fireChannelInactive();
                }
            }
        });
    }
}
 